/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University. All rights reserved.
 * 
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 * 
 * 1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license; 2) All redistributions in binary form
 * must reproduce the above copyright notice, this list of conditions and the
 * disclaimer listed in this license in the documentation and/or other materials
 * provided with the distribution; 3) Any documentation included with all
 * redistributions must include the following acknowledgement:
 * 
 * "This product includes software developed by the Community Grids Lab. For
 * further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/."
 * 
 * Alternatively, this acknowledgement may appear in the software itself, and
 * wherever such third-party acknowledgments normally appear.
 * 
 * 4) The name Indiana University or Community Grids Lab or NaradaBrokering,
 * shall not be used to endorse or promote products derived from this software
 * without prior written permission from Indiana University. For written
 * permission, please contact the Advanced Research and Technology Institute
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202. 5) Products
 * derived from this software may not be called NaradaBrokering, nor may Indiana
 * University or Community Grids Lab or NaradaBrokering appear in their name,
 * without prior written permission of ARTI.
 * 
 * 
 * Indiana University provides no reassurances that the source code provided
 * does not infringe the patent or any other intellectual property rights of any
 * other entity. Indiana University disclaims any liability to any recipient for
 * claims brought by any other entity based on infringement of intellectual
 * property rights or otherwise.
 * 
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 * GENERATED USING SOFTWARE.
 */
package cgl.narada.transport.udp;

import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.util.StringTokenizer;

import cgl.narada.protocol.Protocol;
import cgl.narada.transport.TransportDebugFlags;
import cgl.narada.transport.TransportHandler;
import cgl.narada.transport.util.BytesUtil;
import cgl.narada.util.ByteUtilities;

/**
 * The receiver end of the UDP communications. Listens for datagram packets
 * arriving at a socket and performs various operations based on situations
 * arising during communications. These received packets are then encapsulated
 * into a specialized data structure and forwarded to the transport handler
 * implenetation residing in the transport layer.
 * 
 * @author Shrideep Pallickara and modified by Beytullah Yildiz $Revision$
 *         $Date$
 */

public class UDPReceiverThread extends Thread implements TransportDebugFlags {
   private DatagramSocket socket;

   private int port;

   public final static int MAX_PACKET_SIZE = 65507;

   private byte[] buffer = new byte[MAX_PACKET_SIZE]; // 16000

   private long messageTracker = 0;

   private String moduleName = "UDPReceiverThread: ";

   private UDPLinkFactory udpLinkFactory;

   private TransportHandler transportHandler;

   private UDPBigPackageHandler udbbigpackagehandler;

   private UDPPingServer pingServer;

   private UDPPingHandler udppinghandler;

   private boolean keepLooping = true;
   
   static private int udpSocketReceiveBufferSize = 0;
   
   private long maxMemory = Runtime.getRuntime().maxMemory();
   private long memoryTracker = 0;
   
   public UDPReceiverThread(DatagramSocket socket,
         UDPLinkFactory udpLinkFactory, UDPPingHandler udppinghandler) {
      this.socket = socket;
      this.udpLinkFactory = udpLinkFactory;
      transportHandler = udpLinkFactory.getTransportHandler();
      setName("UDPReceiverThread");
      
      udbbigpackagehandler = new UDPBigPackageHandler();
      udbbigpackagehandler.initReceiver(UDPReceiverThread_Debug);

      this.udppinghandler = udppinghandler;
      pingServer = new UDPPingServer();
      pingServer.setDebug(UDPReceiverThread_Debug);
      pingServer.start();

      if (UDPReceiverThread_Debug) {
          System.out.println(moduleName + "Max Memory =" + maxMemory/1000000 
                             + "MB");
      }
      if (udpSocketReceiveBufferSize > 0) {

         /* set the udp socket receive buffer size to a new value. */
         try {
            int bufferSize = socket.getReceiveBufferSize();
            System.out.println(moduleName
                               + "Initial UDPSocket ReceiveBuffer size:"
                               + bufferSize);
            socket.setReceiveBufferSize(udpSocketReceiveBufferSize);
            System.out.println(moduleName
                               + "New UDPSocket ReceiveBuffer size: "
                               + socket.getReceiveBufferSize());
         } catch (Exception e) {
            e.printStackTrace();
         }
      }

   }

   public static void setReceiveBufferSize(int size) {
      udpSocketReceiveBufferSize = size;
   }

   public void shutdownServices() {
       keepLooping = false;
       udppinghandler.shutdownServices();
       pingServer.shutdownServices();
       udbbigpackagehandler.shutdownServices();
   }
   
   public void run() {
      DatagramPacket dp = new DatagramPacket(buffer, buffer.length);
      if (UDPReceiverThread_Debug) {
         System.out.println(moduleName
                            + "waiting for Datagram packets on port "
                            + socket.getLocalPort() + "...");
      }
      StringTokenizer token;
      int dataInfo;
      int dataLength;
      int port;
      String address;
      try {
         while (keepLooping) {
            socket.receive(dp);
            byte[] data = dp.getData();
            port = dp.getPort();
            dataLength = dp.getLength();
            boolean packetIsBrokenUp = false;

            messageTracker++;
            memoryTracker++;
            // HG: =====
            if (data[0] == Protocol.PING) {
               System.out.println("* UDP: Ping");
               byte[] replydata = new byte[9];

               replydata[0] = Protocol.PING_REPLY;
               byte[] timeBytes = BytesUtil
                                           .longToBytes(System
                                                              .currentTimeMillis());
               System.arraycopy(timeBytes, 0, replydata, 1, 8);
               // Simply send the PING header and recieved ID !
               System.arraycopy(data, 1, replydata, 1, 8);

               DatagramPacket pingReply = new DatagramPacket(replydata,
                                                             replydata.length,
                                                             dp.getAddress(),
                                                             dp.getPort());
               socket.send(pingReply);
               continue;
            }

            // ===== !

            if (data[0] == 105) {
               byte[] dataRecv = new byte[dataLength - 1];
               System.arraycopy(data, 1, dataRecv, 0, dataRecv.length);

               long pingtime = ByteUtilities.getLong(dataRecv);

               UDPLink udp_link = udpLinkFactory.getCurrentLink(
                     dp.getAddress(), dp.getPort());

               if (udp_link != null) {
                  String linkname = udp_link.getLinkId();
                  udppinghandler.ackReceived(linkname, pingtime);

               }
               continue;
            }

            if (data[0] == 104) {

               UDPLink udp_link = udpLinkFactory.getCurrentLink(
                     dp.getAddress(), dp.getPort());
               if (UDPReceiverThread_Debug && udp_link != null) {
                  System.out.println(moduleName + "  " + data[0]
                                     + " ping is received for link "
                                     + udp_link.getLinkId());

               }
               try {
                  if (udp_link != null) {
                     pingServer.addPing(udp_link, data);
                  }
               } catch (Exception ex) {
                  System.err
                            .println("Exception in adding ping to the PingSrever vector");
               }

               // create a PingServer class only one server for entire brokering
               // when a ping message received, forward it to this PingServer
               // It should have a method that could respond the ping so that
               // ping sender
               // can understand server is up

               continue;
            }

            if (data[0] == 103) {

               UDPLink udp_link = udpLinkFactory.getCurrentLink(
                     dp.getAddress(), dp.getPort());
               if (UDPReceiverThread_Debug && udp_link != null) {
                  System.out.println(moduleName + "  " + data[0]
                                     + " close messagereceived for link "
                                     + udp_link.getLinkId());
               }

               if (udp_link != null) {
                  udpLinkFactory.reportLinkLoss(udp_link);
                  // transportHandler.manageLinkLoss(udp_link);
                  udppinghandler.removePing(udp_link.getLinkId());
               }
               if (UDPReceiverThread_Debug) {
                  System.out.println("Connection is closed ");
               }
               continue;
            }

            if (data[0] == 102) {
               packetIsBrokenUp = true;
            }

            InetAddress iaInfo = dp.getAddress();
            String hostInfo = iaInfo.getHostAddress();
            int portInfo = dp.getPort();
            String sender = "udp://" + hostInfo + ":" + port;

            if (!packetIsBrokenUp) { 

               int protocolHeader = data[0];

               if (protocolHeader == Protocol.CONN_REQ) {
                  System.out
                            .println(moduleName
                                     + "Received connection request from client at "
                                     + sender);
                  udpLinkFactory.handleInboundUDPLinkRequest(dp.getAddress(),
                        portInfo);
                  continue;
               }

               if (UDPReceiverThread_Debug) {
                  System.out.println(moduleName + "Received ["
                                     + new String(data, 1, dataLength - 1)
                                     + "] from (" + hostInfo + ":" + portInfo
                                     + ")");
               }

               /** end handling performance measurement */
               
               
               byte[] dataRecv = new byte[dataLength];
               System.arraycopy(data, 0, dataRecv, 0, dataLength);
               data = null;
               
               if (memoryTracker > 10000) {
                  
                  int percentUtilized = getPercentageOfMemoryUtilized();
                  if (percentUtilized >= 60) {
                     String report = moduleName + "Message Tracker = " + messageTracker +
                     "\n\tFreeMem = " + Runtime.getRuntime().freeMemory()/1000000 + "MB "+ 
                     "\n\tTotal Memory = " + maxMemory/1000000 + "MB";
                     
                     System.out.println(report);
                           
                     System.gc();
                     memoryTracker = 0;
                  } else {
                     memoryTracker -= 2000;
                  }
               }
               
               transportHandler.dataReceived(dataRecv, sender);
               
            } else {
               byte[] dataRecv = new byte[dataLength - 1];

               System.arraycopy(data, 1, dataRecv, 0, dataLength - 1);
               byte[] returnedbyte = udbbigpackagehandler
                                                         .receivedPackages(dataRecv);
               if (returnedbyte != null) {
                  transportHandler.dataReceived(returnedbyte, sender);
               }
            } // end of else

         }
         /** end while(true) */
      } catch (SocketException se) {
          if (keepLooping) {
              se.printStackTrace();
          }
      } catch (java.io.IOException ioe) {
         ioe.printStackTrace();
      }
      /* end try-catch */
      if (UDPReceiverThread_Debug) {
          System.out.println(moduleName + "Shutting down ...");
      }
   }
   /* end (run) */
   
   
   public int getPercentageOfMemoryUtilized() {
      long memoryCurrentlyUtilized = Runtime.getRuntime().totalMemory();
      
      double ratio = ((double)memoryCurrentlyUtilized/(double)maxMemory);
      double percent = ratio * (double)100;
      
      return (int)percent;
      
   }
}
